package com.jddup.program.dtm.mysql;

import com.jddup.common.pojo.TableColumn;
import com.jddup.common.pojo.TableMeta;
import com.jddup.context.JddConfigContext;
import com.jddup.program.dtm.DbToTableMetaConvertor;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by ZZB on 2017/12/4.
 */
public class MysqlDbToTableMetaConvertor implements DbToTableMetaConvertor {
    private static final String MysqlTableMetaSQL =
            "SELECT\n" +
                    "  `COLUMN_NAME`                  AS columnName,\n" +
                    "  UPPER(DATA_TYPE)               AS columnType,\n" +
                    "  CHARACTER_MAXIMUM_LENGTH       AS maxLength,\n" +
                    "  NUMERIC_PRECISION              AS `precision`,\n" +
                    "  NUMERIC_SCALE                  AS scale,\n" +
                    "  (IS_NULLABLE = 'YES')          AS isNullable,\n" +
                    "  (COLUMN_KEY = 'PRI')           AS isPrimaryKey,\n" +
                    "  (COLUMN_TYPE LIKE '%unsigned') AS isSigned,\n" +
                    "  COLUMN_DEFAULT                 AS defaultValue,\n" +
                    "  UPPER(COLUMN_TYPE)                    AS typeFullName\n" +
                    "FROM information_schema.columns\n" +
                    "WHERE table_schema = '{dbName}'\n" +
                    "      AND table_name = '{tableName}';";

    @Override
    public TableMeta convertFromTableName(String tableName, String datasourceAlias) throws SQLException {
        TableMeta tableMeta = new TableMeta();
        tableMeta.setTableName(tableName);
        List<TableColumn> list = new ArrayList<>();

        Connection connection = JddConfigContext.getDataSourceMap().get(datasourceAlias).getConnection();

        if (tableExist(connection, tableName)) {
            String schema = connection.getCatalog();
            String sql = MysqlTableMetaSQL.replace("{tableName}", tableName).replace("{dbName}", schema);
            PreparedStatement ps = connection.prepareStatement(sql);
            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                TableColumn tableColumn = new TableColumn();
                tableColumn.setColumnName(rs.getString("columnName"));
                tableColumn.setColumnType(rs.getString("columnType"));
                tableColumn.setMaxLength(rs.getLong("maxLength"));
                tableColumn.setPrecision(rs.getLong("precision"));
                tableColumn.setScale(rs.getLong("scale"));
                tableColumn.setNullable(rs.getBoolean("isNullable"));
                tableColumn.setPrimaryKey(rs.getBoolean("isPrimaryKey"));
                tableColumn.setSigned(rs.getBoolean("isSigned"));
                tableColumn.setDefaultValue(rs.getString("defaultValue"));
                tableColumn.setTypeFullName(rs.getString("typeFullName"));
                list.add(tableColumn);
                if (tableColumn.getPrimaryKey()) {
                    tableMeta.setPrimaryKeyColumn(tableColumn);
                }
            }
        } else {
            tableMeta.setTableExist(false);
        }

        connection.close();
        tableMeta.setTableColumns(list);
        return tableMeta;
    }

    @Override
    public boolean tableExist(Connection connection, String tableName) {
        try {
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("select 1 from `" + tableName + "` limit 1");
            resultSet.next();
            resultSet.close();
            statement.close();
            return true;
        } catch (SQLException e) {
            return false;
        }
    }
}
